package com.dw

import com.doit.dw.dws.MakeAccPageWideJob.followPvPlus
import com.doit.dw.dws.{PageBean, PageNode}
import org.apache.spark.sql.SparkSession

import javax.validation.Path.BeanNode
import scala.collection.mutable.ListBuffer

/**
 * @Date: 23.6.10 
 * @Author: Hang.Nian.YY
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 */
object TestTreeLogic {
  def main(args: Array[String]): Unit = {
    System.setProperty("HADOOP_USER_NAME", "root")
    val session = SparkSession.builder()
      .master("local[*]")
      .appName(this.getClass.getSimpleName)
      .enableHiveSupport()
      .getOrCreate()
    val data = session.sql(
      """
        | select *
        |      from doe39.test_tree
        |""".stripMargin).rdd.map(row => {
      val guid = row.getAs[Long]("guid")
      val session_id = row.getAs[String]("session_id")
      val page_url = row.getAs[String]("page_url")
      val ref_url = row.getAs[String]("ref_url")
      val ts = row.getAs[Long]("ts")
      Bean(guid, session_id, page_url, ref_url, ts)
    })
    val grouped = data.groupBy(_.session_id)
    val roots = grouped.map(tp => {
      val sorted = tp._2.toList.sortBy(_.ts)
      var seqs: ListBuffer[BeanNode] = null
      for (elem <- sorted) {
        val currentNode = BeanNode(elem, ListBuffer[BeanNode]())
        if (seqs == null) {
          seqs = new ListBuffer[BeanNode]()
          seqs.append(currentNode)
        } else {
          seqs.append(currentNode)
          val maybeNode: Option[BeanNode] = seqs.reverse.find(b => b.elem.page_url.equals(currentNode.elem.ref_url))
          if (maybeNode.isDefined) {
            maybeNode.get.nodesToNodes.append(currentNode)
          }
        }
      }
      seqs.head
    })

    val res= roots.flatMap(root => { // 一个会话中所有页面组成的根节点
      val midRes = ListBuffer[(Bean, Int)]()
      followPvPlus(root, midRes)
      midRes
    })
    res.foreach(println)

    session.close()

  }


  def followPvPlus(node: BeanNode , midRes:ListBuffer[(Bean,Int)]): Int = {
    var pv: Int = 0

    val children = node.nodesToNodes
    if(children!=null && children.size>0){
      pv = children.size
      for (elem <- children) {
        var  url = elem.elem
        pv += followPvPlus(elem ,midRes)
      }
    }
    midRes.append((node.elem,pv))
    pv
  }

}
